package com.knight.hadoop.day12.zuoye;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


/**
 * 日志解析第二步
 * @author 
 *
 */
public class LogBeanSecondStep {
	static NullWritable v = NullWritable.get();
	static class LogBeanSecondStepMap extends Mapper<LongWritable, Text, LogBean, LogBean>{
		
		//输入数据：
		//2013-09-19 01:57:371.202.186.37sessionId/nodejs-async-windjs/https://www.google.com.tw/00
		static LogBean logBean = new LogBean();
		
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			String[] splitStr = line.split("\001");
			String strValid = splitStr[0];
			String ip = splitStr[1];
			String remoteUser = splitStr[2];
			String status = splitStr[3];
			String bodyBytesSend = splitStr[4];
			String date = splitStr[5];
			String requestURL = splitStr[6];
			String referal = splitStr[7];
			String userAgent = splitStr[8];

			
			logBean.set(ip, remoteUser, status, bodyBytesSend, date, requestURL, referal, userAgent);
			logBean.setValid(new Boolean(strValid));
			context.write(logBean, logBean);
		}
	}
	
	static class LogBeanSecondStepReducer extends Reducer<LogBean, LogBean, Text, NullWritable>{
		
		static Text k = new Text();
		@Override
		protected void reduce(LogBean key, Iterable<LogBean> values,Context context)
				throws IOException, InterruptedException {
			
			List<LogBean2> listBean = new ArrayList<>();
			for (LogBean bean : values) {
				LogBean2 logBean2 = new LogBean2();
				logBean2.set(bean);
				listBean.add(logBean2);
			}
			
			
			String uuid = UUID.randomUUID().toString();
			
			for (int i = 0; i < listBean.size(); i++) {
//				LogBean bean = listBean.get(i);
//
//				if (i > 0) {
//					LogBean preBean = listBean.get(i - 1);
//					String date = bean.getDate();
//					String preDate = preBean.getDate();
//					if (Utils.compareDate(date, preDate)) {
//						bean.setSessionId(UUID.randomUUID().toString());
//						bean.setStepNum(1);
//						bean.setStayTime(10);
//					} else {
//						bean.setSessionId(preBean.getSessionId());
//						bean.setStepNum(preBean.getStepNum()+1);
//						bean.setStayTime(Utils.compareDateSec(date, preDate));
//					}
//				} else {
//					bean.setSessionId(uuid);
//					bean.setStepNum(1);
//					bean.setStayTime(10);
//				}
//				
//				k.set(bean.toString());
//				context.write(k, v);
			}
		}
	}

	public static void main(String[] args) throws Exception {
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);
		job.setJarByClass(LogBeanSecondStep.class);
		job.setMapperClass(LogBeanSecondStepMap.class);
		job.setReducerClass(LogBeanSecondStepReducer.class);
		job.setGroupingComparatorClass(GroupLogComparator.class);

		job.setMapOutputKeyClass(LogBean.class);
		job.setMapOutputValueClass(LogBean.class);
		
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(NullWritable.class);
		
		FileInputFormat.setInputPaths(job, new Path("E:\\home\\webLog\\output"));
		FileOutputFormat.setOutputPath(job, new Path("E:\\home\\webLog\\step2\\output"));

		boolean res = job.waitForCompletion(true);

		System.exit(res ? 0 : 1);
	}
}
